package com.lmh.source;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.lmh.utils.DateBaseUtil;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDrivenSource;
import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.source.AbstractSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.*;


/**
 * @program flume-component
 * @Description 采集数据库数据
 * @Author limeihua
 * @Date 2024/09/02 15:45
 * @Version 1.0
 */
public class JdbcSource extends AbstractSource implements Configurable, EventDrivenSource {
    private static final Logger logger = LoggerFactory.getLogger(JdbcSource.class);
    private String username;
    private String password;
    private String database;
    private String queryFields;
    private Connection connection;
    private String url;
    private String tableName;
    private String separator;

    @Override
    public void configure(Context context) {
        url = context.getString("url", "jdbc:mysql://localhost:3306/");
        username = context.getString("user", "root");
        password = context.getString("password", "159357");
        database = context.getString("database", "test");
        tableName = context.getString("tableName");
        // 采集查询字段
        queryFields = context.getString("queryFields", "*");
        // 每隔数值之间的分隔符
        separator = context.getString("separator", "\t");
    }

    @Override
    public synchronized void start() {
        try {
            ChannelProcessor channelProcessor = getChannelProcessor();
            connection = DriverManager.getConnection(url + database, username, password);
            Statement CStatement = connection.createStatement();
            ResultSet resultSetCount = CStatement.executeQuery("select count(*) count_row from " + tableName);
            resultSetCount.next();
            int rate = 0;
            int countRow = resultSetCount.getInt("count_row");
            if (countRow <= 500) {
                DateBaseUtil.paginationExecute("select " + queryFields + "from " + tableName, url + database, username, password, channelProcessor, separator);
            } else {
                while (countRow > 0) {
                    DateBaseUtil.paginationExecute("select " + queryFields + "from " + tableName + " limit " + rate * 500 + ",500", url + database, username, password, channelProcessor, separator);
                    countRow = countRow - 500;
                    rate= rate + 1;
                }
            }
        } catch (SQLException e) {
            logger.error("Error executing SQL query", e);
        }
    }

    @Override
    public synchronized void stop() {
        if (connection != null) {
            try {
                connection.close();
            } catch (SQLException e) {
                throw new RuntimeException(e);
            }
        }
    }
}


